package com.fwmagic.flink.state

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object MapWithStateByScala {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.enableCheckpointing(5000)

    val dataStream: DataStream[String] = env.socketTextStream("localhost", 8888)

    val wordAndOne: DataStream[(String, Int)] = dataStream
      .flatMap(_.split("\\s"))
      .map(word => {
        if (word.equals("stop")){
          throw new RuntimeException("===>程序挂了！")
        }
        (word, 1)
      })

    val sumed: DataStream[(String, Int)] = wordAndOne.keyBy(0).mapWithState((input: (String, Int), state: Option[Int]) => {
      val word: String = input._1
      val count: Int = input._2

      state match {
        case Some(value) => {
          val total = count + value
          ((word, total), Some(total))
        }
        case None => {
          (input, Some(input._2))
        }
      }
    })

    sumed.print()

    env.execute("MapWithStateByScala")
  }

}
